package com.frog.queue.starter.sns;


import com.frog.queue.starter.sns.config.SnsEndpointFactory;
import com.frog.queue.starter.sns.enums.Protocol;
import org.springframework.util.Assert;
import software.amazon.awssdk.services.sns.model.*;
import software.amazon.awssdk.utils.StringUtils;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * @author wuqiong 2022/3/3
 * @description frog sns client
 */
public class FrogSnsClient {


    /**
     * 方法描述: 发布一个消息主题消息
     *
     * @param topicArn               主题 arn
     * @param message                消息内容
     * @param messageGroupId         消息组ID, 仅适用于 FIFO 消息
     * @param messageDeduplicationId 消息重复ID, 仅适用于 FIFO 消息
     * @param messageAttributes      消息属性, 可自定义的
     * @return boolean
     * @author wuqiong 2022/3/4 15:54
     */
    public static Boolean publishTopic(
            String clientName, String topicArn, String message, String messageGroupId, String messageDeduplicationId,
            Map<String, MessageAttributeValue> messageAttributes) throws SnsException {
        Assert.isTrue(FrogQueueUtils.checkTopicArn(topicArn), "参数 topicArn 值不正确,请检查");
        Assert.hasText(message, "参数 message 值不能为空,请检查");
        PublishRequest.Builder publishBuilder = PublishRequest.builder().topicArn(topicArn).message(message);
        if (StringUtils.isNotBlank(messageGroupId)) publishBuilder.messageGroupId(messageGroupId);
        if (StringUtils.isNotBlank(messageDeduplicationId))
            publishBuilder.messageDeduplicationId(messageDeduplicationId);
        if (messageAttributes != null && !messageAttributes.isEmpty())
            publishBuilder.messageAttributes(messageAttributes);
        return SnsEndpointFactory.getSnsClient(clientName).publish(publishBuilder.build()).sdkHttpResponse().isSuccessful();
    }

    public static Boolean publishTopic(String topicArn, String message) throws SnsException {
        return publishTopic(null, topicArn, message, null, null, null);
    }

    public static Boolean publishTopic(String clientName, String topicArn, String message) throws SnsException {
        return publishTopic(clientName, topicArn, message, null, null, null);
    }

    /**
     * 方法描述: 创建一个订阅
     *
     * @param protocol 协议
     * @param topicArn 主题 arn
     * @param endpoint 端点
     * @return String
     * @author wuqiong 2022/3/4 16:01
     */
    public static String createSubscribe(String clientName, Protocol protocol, String topicArn, String endpoint) throws SnsException {
        Assert.isTrue(FrogQueueUtils.checkTopicArn(topicArn), "参数 topicArn 值不正确,请检查");
        Assert.hasText(endpoint, "参数 endpoint 值不能为空,请检查");
        SubscribeRequest request = SubscribeRequest.builder()
                .protocol(protocol.getName())
                .topicArn(topicArn)
                .endpoint(endpoint)
                .returnSubscriptionArn(true)
                .build();
        return SnsEndpointFactory.getSnsClient(clientName).subscribe(request).subscriptionArn();
    }
    public static String createSubscribe(Protocol protocol, String topicArn, String endpoint) throws SnsException {
        return createSubscribe(null, protocol, topicArn, endpoint);
    }


    /**
     * 方法描述: 取消订阅
     *
     * @param topicArn 主题 arn
     * @param endpoint 端点
     * @return boolean
     * @author wuqiong 2022/3/4 16:03
     */
    public static boolean unSubscribe(String clientName, String topicArn, String endpoint) throws SnsException {
        Assert.isTrue(FrogQueueUtils.checkTopicArn(topicArn), "参数 topicArn 值不正确,请检查");
        Assert.hasText(endpoint, "参数 endpoint 值不能为空,请检查");

        // 查找对应订阅
        String subscriptionArn = getTopicArnValue(clientName, topicArn, endpoint);
        if (StringUtils.isNotBlank(subscriptionArn)) {
            // Assert.isTrue(false,"在 " + topicArn + " 主题下未查询到 endpoint: " + endpoint + " 的订阅信息,因此无法删除");
            return false;
        }
        UnsubscribeRequest request = UnsubscribeRequest.builder()
                .subscriptionArn(subscriptionArn)
                .build();
        return SnsEndpointFactory.getSnsClient(clientName).unsubscribe(request).sdkHttpResponse().isSuccessful();

    }

    /**
     * 方法描述:基于给定主题和端点，返回订阅的 arn
     *
     * @param topicArn 主题 arn
     * @param endpoint 端点
     * @return String
     * @author wuqiong 2022/3/4 16:13
     */
    public static String getTopicArnValue(String clientName, String topicArn, String endpoint) throws SnsException {
        Assert.isTrue(FrogQueueUtils.checkTopicArn(topicArn), "参数 topicArn 值不正确,请检查");
        Assert.hasText(endpoint, "参数 endpoint 值不能为空,请检查");
        ListSubscriptionsByTopicRequest request = ListSubscriptionsByTopicRequest.builder().topicArn(topicArn).build();
        ListSubscriptionsByTopicResponse result = null;
        do {
            if (result != null) {
                request = ListSubscriptionsByTopicRequest.builder().topicArn(topicArn).nextToken(result.nextToken()).build();
            }
            result = SnsEndpointFactory.getSnsClient(clientName).listSubscriptionsByTopic(request);
            for (Subscription sub : result.subscriptions()) {
                if (sub.endpoint().compareTo(endpoint) == 0) return sub.subscriptionArn();
            }
        } while (result.nextToken() != null);
        return null;
    }

    /**
     * 方法描述: 返回所有的topic arn
     *
     * @return List<String>
     * @author wuqiong 2022/3/4 16:31
     */
    public static List<String> getTopicArnList(String clientName) {
        return SnsEndpointFactory.getSnsClient(clientName).listTopics().topics().stream().map(e -> e.topicArn()).collect(Collectors.toList());
    }

    /**
     * <p> 方法描述: 返回当前主题的所有 endpoint 和 subscriptionArn </p>
     * <p> 其中 map 是一对组合值 </p>
     * <p> key = endpoint </p>
     * <p> value = subscriptionArn </p>
     *
     * @param topicArn 主题 arn
     * @return List
     * @author wuqiong 2022/3/8 13:08
     */
    public static Map<String, String> getTopicArnAllEndpointAndSubscription(String clientName, String topicArn) throws SnsException {
        Assert.isTrue(FrogQueueUtils.checkTopicArn(topicArn), "参数 topicArn 值不正确,请检查");
        ListSubscriptionsByTopicRequest request = ListSubscriptionsByTopicRequest.builder().topicArn(topicArn).build();
        ListSubscriptionsByTopicResponse result = null;
        Map<String, String> resultEndpoints = new HashMap<>();
        do {
            if (result != null) {
                request = ListSubscriptionsByTopicRequest.builder().topicArn(topicArn).nextToken(result.nextToken()).build();
            }
            result = SnsEndpointFactory.getSnsClient(clientName).listSubscriptionsByTopic(request);
            if (result.hasSubscriptions()) {
                result.subscriptions().forEach(e -> resultEndpoints.put(e.endpoint(), e.subscriptionArn()));
            }
        } while (result.nextToken() != null);
        return resultEndpoints;
    }

}
